/**
 * Copyright 2009-2010 by The Regents of the University of California
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * you may obtain a copy of the License from
 * 
 *     http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package edu.ucr.cs.dblab.acestore.base;

import java.io.IOException;
import java.nio.ByteBuffer;

/**
 * A page adapter for pages in linked-list format.
 * 
 * A page is organized as:
 * 
 * +----------+----------+-----+------------+-------+---------------+------------------+------------+
 * | Record 0 | Record 1 | ... | Record n-1 | blank | Next Page:int | Record Count:int | Reserved=0 |
 * +----------+----------+-----+------------+-------+---------------+------------------+------------+
 * 
 * And each record is in the following format:
 * 
 * +------------------+--------------------+--------------------+-----+--------------------+--------+--------+-----+--------+
 * | FieldCount:short | Field0EndOfs:short | Field1EndOfs:short | ... | FieldnEndOfs:short | Field0 | Field1 | ... | Fieldn |
 * +------------------+--------------------+--------------------+-----+--------------------+--------+--------+-----+--------+
 * 
 * Next page and record count are both 4-byte integers.
 * 
 * @author jarod
 * @created Jan 10, 2011
 */
public class LinkListPageAdapter implements IPageAdapter {

	private ByteBuffer page;
	
	private static final int RESERVED = 0;

	public LinkListPageAdapter(ByteBuffer page) {
		this.page = page;
	}

	public int getNextPage() {
		return page.getInt(page.capacity() - RESERVED - 2 * Integer.SIZE);
	}

	public void setNextPage(int nextPage) {
		page.putInt(page.capacity() - RESERVED - 2 * Integer.SIZE, nextPage);
	}

	private void setRecordCount(int recordCount) {
		page.putInt(page.capacity() - RESERVED - 1 * Integer.SIZE, recordCount);
	}

	/**
	 * Get the offset of the record with given index.
	 * 
	 * @param recordIdx
	 * @return
	 */
	private int getRecordEndOffset(int recordIdx) {
		int recordCount = getRecordCount();
		if (recordIdx >= recordCount || recordIdx < 0) {
			return -1;
		}else{
			int offset = -1;
			for(int i = 0; i <= recordIdx; i++){
				offset += 1;
				offset += Short.SIZE / 8;
				offset += page.getShort(offset);
			}
			return offset;
		}
	}

	/**
	 * Delete a record from this page.
	 * 
	 * When deleting a record, records after this one will be moved ahead
	 * so that all empty space is in a single block. 
	 * 
	 * @see edu.ucr.cs.dblab.acestore.base.IPageAdapter#delete(int)
	 */
	@Override
	public boolean delete(int recordIdx) throws IOException {
		int start = getRecordEndOffset(recordIdx);
		if (start < 0) {
			System.err.println("Cannot find the record to be deleted!");
			return false;
		}
		// Delete the record
		int recordCount = getRecordCount();
		if (recordIdx < recordCount - 1) {
			// Adjust bytes to cover the blank space (reallocate records)
			int end = getRecordEndOffset(recordCount - 1);
			int copyTo = (recordIdx == 0 ? 0
					: getRecordEndOffset(recordIdx - 1));
			byte[] buf = new byte[end - start];
			page.get(buf, start, end - start);
			page.put(buf, copyTo, end - start);
		}
		setRecordCount(recordCount - 1);
		return true;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see edu.ucr.cs.dblab.acestore.base.IPageAdapter#getRecordCount()
	 */
	@Override
	public int getRecordCount() {
		return page.getInt(page.capacity() - RESERVED - 1 * Integer.SIZE);
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see edu.ucr.cs.dblab.acestore.base.IPageAdapter#insert(byte[])
	 */
	@Override
	public boolean insert(byte[] record) throws IOException {
		if(record.length <=0){
			System.err.println("Try to insert a empty record!");
			return true;
		}
		int recordCount = getRecordCount();
		int newOffset = (recordCount == 0 ? 0 : (getRecordEndOffset(recordCount - 1) + 1));
		if( page.capacity() 
				- newOffset 				// Start from new offset
				- RESERVED 					// Remove reversed space
				- Integer.SIZE / 8 * 2 		// Remote record count and next page field
				< record.length){
			// Not enough space for the new record
			return false;
		}
		// Insert data
		page.put(record, newOffset, record.length);
		setRecordCount(recordCount + 1);
		return true;
	}

	/**
	 * 
	 * @param offset
	 * @param length
	 * @return
	 * @throws IOException
	 */
	private ByteBuffer read(int offset, int length) throws IOException {
		ByteBuffer data = ByteBuffer.wrap(page.array(), offset, length);
		return data;
	}

	/**
	 * Read a record from the page, by the given record index.
	 * 
	 * If no data is available for such a record index, a null is returned.
	 * 
	 * @see edu.ucr.cs.dblab.acestore.base.IPageAdapter#read(int)
	 */
	@Override
	public ByteBuffer read(int recordIdx) throws IOException {
		int offset = (recordIdx == 0 ? 0 : getRecordEndOffset(recordIdx - 1));
		int end = getRecordEndOffset(recordIdx);
		if(end > offset){
			return read(offset, end - offset + 1);
		}
		return null;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see edu.ucr.cs.dblab.acestore.base.IPageAdapter#update(byte[], int)
	 */
	@Override
	public boolean update(byte[] record, int recordIdx) throws IOException {
		int offset = (recordIdx == 0 ? 0 : getRecordEndOffset(recordIdx - 1));
		int end = getRecordEndOffset(recordIdx);
		if(end - offset + 1 == record.length){
			page.put(record, offset, end - offset + 1);
			return true;
		}else{
			// If after update, the size of the record is changed, the original
			// record is deleted, and the new record is added then.
			delete(recordIdx);
			return insert(record);
		}
	}

	@Override
	public void reset(ByteBuffer page) {
		this.page = page;
	}

	@Override
	public ByteBuffer getPage() {
		return this.page;
	}

	@Override
	public void initPage(ByteBuffer page) {
		this.setNextPage(-1);
		this.setRecordCount(0);
	}

	@Override
	public void writeReserved(byte[] head) throws IOException {
		if(head.length > RESERVED)
			throw new IOException("Try to write a too long reserved field: (" + RESERVED + ":" + head.length + ")");
		page.put(head, page.capacity() - RESERVED, head.length);
	}

	@Override
	public int getReservedLength() {
		return RESERVED;
	}

}
